package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdInteractionReview
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/12 11:42
 * @version: 1.0
 */

public class DwdInteractionReview extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdInteractionReview().start(Constant.TOPIC_DWD_TRADE_INTERACTION_REVIEW,Constant.TOPIC_DWD_TRADE_INTERACTION_REVIEW,4,10018);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
       readOdsDb(tableEnv,groupId);
        Table table = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id ,\n" +
                "    `data`['create_time'] create_time ,\n" +
                "    `data`['update_time'] update_time ,\n" +
                "    `data`['rating'] rating ,\n" +
                "    `data`['review'] review ,\n" +
                "    `data`['status'] status ,\n" +
                "    `data`['doctor_id'] doctor_id ,\n" +
                "    `data`['patient_id'] patient_id ,\n" +
                "    `data`['user_id'] user_id\n" +
                "from topic_db_medical\n" +
                "where `table` = 'consultation'\n" +
                "  and `type` = 'update'\n" +
                "  and `data`['status'] = '207'");

        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_INTERACTION_REVIEW+"(\n" +
                "    id string,\n" +
                "    create_time string,\n" +
                "    update_time string,\n" +
                "    rating string,\n" +
                "    review string,\n" +
                "    status string,\n" +
                "    doctor_id string,\n" +
                "    patient_id string,\n" +
                "    user_id string\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_INTERACTION_REVIEW));

        table.executeInsert(Constant.TOPIC_DWD_TRADE_INTERACTION_REVIEW);

    }
}
